package com.wfm;

import java.util.Map;
import org.apache.kafka.clients.producer.*;

import java.util.Properties;
import java.util.concurrent.ExecutionException;
import org.apache.kafka.common.Cluster;

public class Product {

    public static void main(String[] args) throws ExecutionException, InterruptedException {
        // 配置
        Properties props = configKfk();
        // 发送消息
        sendAsyncPartition(props);


    }

    private static Properties configKfk() {
        // 1. 创建 kafka 生产者的配置对象
        Properties props = new Properties();
        // 2. 给 kafka 配置对象添加配置信息:bootstrap.servers
        props.put("bootstrap.servers", "110.40.196.180:9192");
        // 被发送到broker的任何消息的格式都必须是字节数组
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        // 非必须参数配置
        // acks=0表明producer完全不管发送结果；
        // acks=all或-1表明producer会等待ISR所有节点均写入后的响应结果；
        // acks=1，表明producer会等待leader写入后的响应结果
//        props.put("acks", "-1");
//        // 发生可重试异常时的重试次数
//        props.put("retries", 3);
//        // producer会将发往同一分区的多条消息封装进一个batch中，
//        // 当batch满了的时候，发送其中的所有消息,不过并不总是等待batch满了才发送消息；
//        props.put("batch.size", 323840);
//        // 控制消息发送延时，默认为0，即立即发送，无需关心batch是否已被填满。
//        props.put("linger.ms", 10);
//        // 指定了producer用于缓存消息的缓冲区大小，单位字节，默认32MB
//        // producer启动时会首先创建一块内存缓冲区用于保存待发送的消息，然后由另一个专属线程负责从缓冲区中读取消息执行真正的发送
//        props.put("buffer.memory", 33554432);
//        // 设置producer能发送的最大消息大小
//        props.put("max.request.size", 10485760);
//        // 设置是否压缩消息，默认none
//        props.put("compression.type", "lz4");
//        // 设置消息发送后，等待响应的最大时间
//        props.put("request.timeout.ms", 30);
        return props;
    }

    // 同步发送
    private static void sendSync(Properties props)
            throws ExecutionException, InterruptedException {
        // 创建kfk生产者对象
        Producer<String, String> producer = new KafkaProducer<String, String>(props);
        for (int i = 5; i < 10; i++) {
            RecordMetadata recordMetadata = producer.send(
                    new ProducerRecord<>("my-first-topic", "key" + i, "value" + i)).get();
            System.out.println("===生产者==主题"+recordMetadata.topic()+
                    "分区"+recordMetadata.partition()+"长度"+recordMetadata.serializedValueSize());
        }
        //  关闭资源
        producer.close();
    }


    // 普通异步发送
    private static void sendComAsync(Properties props) {
        // 创建kfk生产者对象
        Producer<String, String> producer = new KafkaProducer<String, String>(props);

        for (int i = 5; i < 10; i++) {
            producer.send(new ProducerRecord<>("my-first-topic", "key" + i, "value" + i));
        }

        //  关闭资源
        producer.close();
    }

    // 异步发送消息
    private static void sendAsync(Properties props) throws InterruptedException {
        // 创建kfk生产者对象
        Producer<String, String> producer = new KafkaProducer<String, String>(props);

        for (int i = 5; i < 10; i++) {
            producer.send(new ProducerRecord<>("my-first-topic", "key" + i, "value" + i), new Callback() {
                // 该方法在Producer收到ack时调用，为异步调用
                @Override
                public void onCompletion(RecordMetadata recordMetadata, Exception e) {
                    if (e == null) {
                        // 没有异常,输出信息到控制台
                        System.out.println(" 主 题 : " + recordMetadata.topic() + "->" + "分区:" + recordMetadata.partition());
                    } else {
                        // 出现异常打印
                        e.printStackTrace();
                    }
                }
            });

            // 延迟一会会看到数据发往不同分区
            Thread.sleep(2);
        }

        //  关闭资源
        producer.close();
    }

    // 异步发送消息,指定分区1,分区必须要存在，如果指定了分区,根据key的hash应该会失效
    private static void sendAsyncPartition(Properties props) throws InterruptedException {
        // 创建kfk生产者对象
        Producer<String, String> producer = new KafkaProducer<String, String>(props);

        for (int i = 5; i < 10; i++) {
            producer.send(new ProducerRecord<>("my-first-topic",0,"", "partition-value" + i), new Callback() {
                // 该方法在Producer收到ack时调用，为异步调用
                @Override
                public void onCompletion(RecordMetadata recordMetadata, Exception e) {
                    if (e == null) {
                        // 没有异常,输出信息到控制台
                        System.out.println(" 主 题 : " + recordMetadata.topic() + "->" + "分区:" + recordMetadata.partition());
                    } else {
                        // 出现异常打印
                        e.printStackTrace();
                    }
                }
            });

            // 延迟一会会看到数据发往不同分区
            Thread.sleep(2);
        }

        //  关闭资源
        producer.close();
    }

    // 自定义分区
    private static void sendAsyncPartitionSelf(Properties props) throws InterruptedException {
        // 添加自定义分区器
        props.put(ProducerConfig.PARTITIONER_CLASS_CONFIG,"com.wfm.partition.MyPartitioner");

        Producer<String, String> producer = new KafkaProducer<String, String>(props);

        for (int i = 5; i < 10; i++) {
            producer.send(new ProducerRecord<>("my-first-topic", "atguigu" + i), new Callback() {
                // 该方法在Producer收到ack时调用，为异步调用
                @Override
                public void onCompletion(RecordMetadata recordMetadata, Exception e) {
                    if (e == null) {
                        // 没有异常,输出信息到控制台
                        System.out.println(" 主 题 : " + recordMetadata.topic() + "->" + "分区:" + recordMetadata.partition());
                    } else {
                        // 出现异常打印
                        e.printStackTrace();
                    }
                }
            });

            // 延迟一会会看到数据发往不同分区
            Thread.sleep(2);
        }

        //  关闭资源
        producer.close();
    }

}
